package com.tao.mvpbaselibrary.basic.rxbus;

import androidx.annotation.NonNull;

import android.util.Log;

import com.tao.logger.log.Logger;

import java.util.HashMap;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

public class RxBus {
    private static final String TAG = "RxBus";
    private HashMap<String, CompositeDisposable> mSubscriptionMap;
    private HashMap<String, Disposable> mDisposeMap;
    private static volatile RxBus mRxBus;
    private final Subject<Object> mSubject;

    //单列模式
    public static RxBus getInstanceBus() {
        if (mRxBus == null) {
            synchronized (RxBus.class) {
                if (mRxBus == null) {
                    mRxBus = new RxBus();
                }
            }
        }
        return mRxBus;
    }

    public RxBus() {
        mSubject = PublishSubject.create().toSerialized();
    }


    public static void post(Object o) {
        getBus().postt(o);
    }

    public void postt(Object o) {
        mSubject.onNext(o);
    }

    /**
     * 返回指定类型的带背压的Flowable实例
     *
     * @param <T>
     * @param type
     * @return
     */
    public <T> Flowable<T> getObservable(Class<T> type) {
        return mSubject.toFlowable(BackpressureStrategy.BUFFER)
                .ofType(type);
    }

    /**
     * 一个默认的订阅方法
     *
     * @param <T>
     * @param type
     * @param next
     * @param error
     * @return
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<T> next, Consumer<Throwable> error) {
        return getObservable(type)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next, error);
    }


    public <T> Disposable doSubscribe(Class<T> type, BaseRxubsConsumer consumer) {
        return doSubscribe(type, consumer.next(), consumer.error());
    }

    /**
     * 是否已有观察者订阅
     *
     * @return
     */
    public boolean hasObservers() {
        return mSubject.hasObservers();
    }

    /**
     * 保存订阅后的disposable
     *
     * @param o
     * @param disposable
     */
    public void addSubscription(Object o, Disposable disposable) {
        if (mSubscriptionMap == null) {
            mSubscriptionMap = new HashMap<>();
        }
        String key = o.getClass().getName();
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).add(disposable);
        } else {
            //一次性容器,可以持有多个并提供 添加和移除。
            CompositeDisposable disposables = new CompositeDisposable();
            disposables.add(disposable);
            mSubscriptionMap.put(key, disposables);
        }
    }

    /**
     * 取消订阅
     *
     * @param o
     */
    public static void unSubscribet(Object o) {
        getBus().unSubscribe(o);
    }

    public void unSubscribe(Object o) {
        if (mSubscriptionMap == null) {
            return;
        }

        String key = o.getClass().getName();
        if (!mSubscriptionMap.containsKey(key)) {
            return;
        }
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).dispose();
        }

        mSubscriptionMap.remove(key);
    }

    /**
     * 注册
     *
     * @param eventType
     * @param action
     * @param <T>
     */
    public <T> void registerRxBus(Object key, Class<T> eventType, Consumer<T> action) {
        Disposable disposable = doSubscribe(eventType, action, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {
                Logger.e(TAG, throwable);
            }
        });
        addSubscription(key, disposable);
    }

    public static RxBus getBus() {
        return getInstanceBus();
    }

    /**
     * 一个默认的订阅方法
     *
     * @param <T>
     * @param type
     * @param next
     * @param error
     * @return
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<T> next, boolean observerNewTherad, Consumer<Throwable> error) {
        return getObservable(type)
                .subscribeOn(Schedulers.io())
                .observeOn(observerNewTherad ? Schedulers.io() : AndroidSchedulers.mainThread())
                .subscribe(next, error);
    }

    /**
     * 注册
     *
     * @param eventType
     * @param action
     * @param <T>
     */
    public <T> void registerRxBus(Object key, boolean observerNewTherad, Class<T> eventType, Consumer<T> action) {
        Disposable disposable = doSubscribe(eventType, action, observerNewTherad, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {
                Log.e(TAG, throwable.getMessage());
            }
        });
        addSubscription(key, disposable);
    }

    public <T>  void registerBus(Object key, Class<T> eventType, Consumer<T> action) {
        registerRxBus(key, true, eventType, action);
    }

    public <T> void registerBusOnUi(Object key, Class<T> eventType, Consumer<T> action) {
        registerRxBus(key, false, eventType, action);
    }
}